package io.renren.modules.delay;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class DelayQueueManager implements CommandLineRunner {

  private DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
  @Autowired
  private ReportTaskRunner meterTaskRunner;
  @Autowired
  private NotifyResponseTimeoutRunner notifyResponseTimeoutRunner;
  /**
   * 加入到延时队列中
   * @param task
   */
  public void put(DelayTask task) {
    log.info("加入延时任务：{}", task);
    delayQueue.put(task);
  }

  /**
   * 取消延时任务
   * @param task
   * @return
   */
  public boolean remove(DelayTask task) {
    log.info("取消延时任务：{}", task);
    return delayQueue.remove(task);
  }

  /**
   * 取消延时任务
   * @param taskid
   * @return
   */
  public boolean remove(String taskid) {
    return remove(new DelayTask(new TaskBase(taskid), 0));
  }

  @Override
  public void run(String... args) throws Exception {
    log.info("初始化延时队列");
    Executors.newSingleThreadExecutor().execute(new Thread(this::excuteThread));
  }

  /**
   * 延时任务执行线程
   */
  private void excuteThread() {
    while (true) {
      try {
        DelayTask task = delayQueue.take();
        processTask(task);
      } catch (InterruptedException e) {
        break;
      }
    }
  }

  /**
   * 内部执行延时任务
   * @param task
   */
  private void processTask(DelayTask task) {
    log.info("执行延时任务：{}", task);
    if (task.getData() instanceof ReportTaskBase) {
      meterTaskRunner.run((ReportTaskBase)task.getData());
    } else if (task.getData() instanceof NotifyResponseTimeoutBase) {
      notifyResponseTimeoutRunner.run((NotifyResponseTimeoutBase)task.getData());
    }
  }
}
